package com.atguigu.flink.charkoer13;

import com.atguigu.flink.been.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.time.ZoneOffset;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.e;

public class FlinkTimeEvent02 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //设置时区
        tEnv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(0));
//毫秒3，秒就是0
        tEnv.executeSql("create table abc(id string,ts bigint,vc int,es as TO_TIMESTAMP_LTZ(ts,0)," +
                "watermark for es as es - interval '3' second" +
                ")with(" +
                " 'connector' = 'filesystem' ," +
                " 'path' = 'input/sensor.txt' ," +
                " 'format' = 'csv' " +
                ")");
        tEnv.sqlQuery("select * from  abc").execute().print();

    }
}
